/*
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or
 * (at your option) any later version.
 * 
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Library General Public License for more details.
 * 
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor Boston, MA 02110-1301,  USA
 */

/**
 * @version thread.c 0.0.1
 * @package Comoro
 * @author Dr.NP <np@bsgroup.org>
 * 
 * Multi-thread operator
 */

#include "bspd.h"

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <pthread.h>

#ifdef HAVE_MALLOC_H
#ifndef __OpenBSD__
#include <malloc.h>
#endif
#endif

extern struct conf setting;
extern struct event_base *ev_base;
struct thread_worker *worker_list[WORKER_POOL_SIZE];
static int curr_thread = 0;

/* Initialization of thread worker list */
int thread_init()
{
	int num_workers;
	int worker_id;
	int i;
	
	memset(worker_list, 0, sizeof(struct thread_worker *) * WORKER_POOL_SIZE);

	num_workers = setting.srv_worker;

	// Initialize main thread
	thread_worker_new(0);

	if (setting.srv_enable_thread)
	{
		if (num_workers < 1)
		{
			num_workers = 1;
		}

		for (i = 0; i < num_workers; i ++)
		{
			worker_id = thread_list_find_free();
			thread_worker_new(worker_id);
		}
	}
	
	return RTN_SUCCESS;
}

/* Generate a new worker thread */
int thread_worker_new(int id)
{
	if (id < 0)
	{
		// Thread pool full
		fprintf(stderr, "Thread pool full, cannot handle more thread\n");
		log_add("Thread pool full");

		exit(EXIT_MEMORY_ERROR);
	}
	
	struct thread_worker *worker = (struct thread_worker *) malloc(sizeof(struct thread_worker));
	int fds[2];

	if (pipe(fds))
	{
		fprintf(stderr, "Notify pipe create error\n");
		log_add("Notify pipe create error");
		
		exit(EXIT_PIPE_ERROR);
	}

	worker->notify_recv_fd = fds[0];
	worker->notify_send_fd = fds[1];

	conn_setnonblock(fds[0]);
	conn_setnonblock(fds[1]);

	if (id > 0)
	{
		worker->ev_base = event_init();

		if (!worker->ev_base)
		{
			fprintf(stderr, "Worker event base initialize error\n");
			log_add("Worker event base initialize error");

			exit(EXIT_GENERAL);
		}

		pthread_attr_t thread_attr;
		int ret;

		pthread_attr_init(&thread_attr);
		ret = pthread_create(&worker->tid, &thread_attr, thread_worker_init, (void *) worker);

		if (ret)
		{
			fprintf(stderr, "Thread create error\n");
			log_add("Thread create error");

			exit(EXIT_THREAD_ERROR);
		}

		fprintf(stderr, "Worker thread %d created\n", id);
		log_add("Worker thread %d created", id);
	}

	else
	{
		// Main thread
		worker->ev_base = ev_base;
		worker->tid = pthread_self();

		event_set(&worker->ev_notify, worker->notify_recv_fd, EV_READ|EV_PERSIST, thread_worker_process, worker);
		event_base_set(worker->ev_base, &worker->ev_notify);
		event_add(&worker->ev_notify, 0);
	}

	worker->id = id;
	worker_list[id] = worker;
	curr_thread = id;

	return RTN_SUCCESS;
}

/* Worker initialize */
void * thread_worker_init(void *arg)
{
	struct thread_worker *worker = (struct thread_worker *) arg;

	event_set(&worker->ev_notify, worker->notify_recv_fd, EV_READ|EV_PERSIST, thread_worker_process, worker);
	event_base_set(worker->ev_base, &worker->ev_notify);
	event_add(&worker->ev_notify, 0);

	event_base_loop(worker->ev_base, 0);
	
	return NULL;
}

/* Transfer a new event to worker */
void thread_worker_process(int fd, short ev, void *arg)
{
	struct thread_worker *worker = (struct thread_worker *) arg;
	char sig_dispatch[5];
	char worker_type;
	int sig_id;
	
	// Read mark from pipe
	while(5 == read(fd, sig_dispatch, 5))
	{
		worker_type = sig_dispatch[0];
		sig_id = (int) get_u_int(sig_dispatch + 1);

		if (WORKER_TYPE_CONN == worker_type)
		{
			struct conn_client *client = conn_client_list_get(sig_id);
			
			event_set(&client->ev_read, client->client_fd, EV_READ|EV_PERSIST, conn_tcp_server_on_data, (void *) client);
			event_base_set(worker->ev_base, &client->ev_read);
			
			if (-1 == event_add(&client->ev_read, NULL))
			{
				fprintf(stderr, "Cannot handle client's data event\n");
			}
		}

		else if (WORKER_TYPE_COIN == worker_type)
		{
			// Nothing for coin server now
		}

		else
		{
			// Unknown worker type
		}
	}

	return;
}

/* Dispatch to worker */
int thread_worker_dispatch(char worker_type, int sig_id, char sel_method)
{
	char sig_dispatch[5];
	int worker_id;
	int res;

	sig_dispatch[0] = worker_type;
	put_u_int ((unsigned int) sig_id, sig_dispatch + 1);
	worker_id = thread_list_find_next();

	if (setting.srv_enable_thread && WORKER_DISPATCH_CHILD_ONLY == sel_method && 0 == worker_id)
	{
		worker_id = thread_list_find_next();
	}

	res = write(worker_list[worker_id]->notify_send_fd, sig_dispatch, 5);

	if (5 != res)
	{
		fprintf(stderr, "Pipe write error\n");
		log_add("Pipe write error");

		return RTN_PIPE_ERROR;
	}

	return RTN_SUCCESS;
}

/* Find a empty slot from worker list */
int thread_list_find_free()
{
	int i;

	for (i = 0; i < WORKER_POOL_SIZE; i ++)
	{
		if (!worker_list[i] || !worker_list[i]->tid)
		{
			// An empty slot
			return i;
		}
	}

	return RTN_GENERAL;
}

/* Find next available worker */
int thread_list_find_next()
{
	int i, tmp;

	for (i = 1; i <= WORKER_POOL_SIZE; i ++)
	{
		tmp = (i + curr_thread) % WORKER_POOL_SIZE;

		if (worker_list[tmp] && worker_list[tmp]->tid)
		{
			curr_thread = tmp;
			return tmp;
		}
	}

	// No available worker???
	return RTN_GENERAL;
}
